package com.wenge.datagroup.storage.service.impl;

import com.alibaba.fastjson.serializer.ValueFilter;

import com.wenge.datagroup.storage.bean.Field;
import com.wenge.datagroup.storage.bean.Topic;
import com.wenge.datagroup.storage.future.AsyncProcessCreater;
import com.wenge.datagroup.storage.service.TransformService;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.apache.spark.broadcast.Broadcast;

import java.io.Serializable;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.concurrent.ConcurrentHashMap;

public class BaseTransformServiceImpl implements TransformService, Serializable {
    public static final Logger logger = Logger.getLogger(BaseTransformServiceImpl.class);
    private static ValueFilter valueFilter = (o, s, o1) -> o1 == null ? "" : o1;


    /**
     * 数据预处理
     *
     * @param orgRecord
     * @param topic
     */
    @Override
    public void transferData(
            JSONObject orgRecord,
            Topic topic,
            Broadcast<AsyncProcessCreater> asyncBro) {
        long start = System.currentTimeMillis();
        try {
            LinkedList<Field> fields = topic.getFields();

            ConcurrentHashMap<String, Object> map = asyncBro.value().asyncProcess().runAsync(
                    fields,
                    orgRecord);
            /*logger.info("orgRecord:"+orgRecord.toJSONString());
            logger.info("map:"+map);
            logger.info("单条数据处理耗时： " + (System.currentTimeMillis() - start) + " ms");
            */
            logger.info("orgRecord："+JSONObject.toJSONString(orgRecord));

            orgRecord.clear();
            orgRecord.putAll(map);
            boolean contzh1 = orgRecord.containsKey("contzh");
            if (contzh1) {
                String contzh = orgRecord.getString("contzh");
                if (StringUtils.isBlank(contzh)||isBlankString(contzh)) {
                    if (orgRecord.containsKey("cont")) {
                        orgRecord.put("contzh", orgRecord.getString("cont"));
                    }
                }
            } else {
                if (orgRecord.containsKey("cont")) {
                    orgRecord.put("contzh", orgRecord.getString("cont"));
                }
            }

            if (orgRecord.containsKey("contentzh")) {
                String titlezh = orgRecord.getString("contentzh");
                if (StringUtils.isBlank(titlezh)||isBlankString(titlezh)) {
                    if (orgRecord.containsKey("content")) {
                        orgRecord.put("contentzh", orgRecord.getString("content"));
                    }
                }
            } else {
                if (orgRecord.containsKey("content")) {
                    orgRecord.put("contentzh", orgRecord.getString("content"));
                }
            }
            if (orgRecord.containsKey("titlezh")) {
                String titlezh = orgRecord.getString("titlezh");
                if (StringUtils.isBlank(titlezh)||isBlankString(titlezh)) {
                    if (orgRecord.containsKey("title")) {
                        orgRecord.put("titlezh", orgRecord.getString("title"));
                    }
                }
            } else {
                if (orgRecord.containsKey("title")) {
                    orgRecord.put("titlezh", orgRecord.getString("title"));
                }
            }

            if (orgRecord.containsKey("readmezh")) {
                String titlezh = orgRecord.getString("readmezh");
                if (StringUtils.isBlank(titlezh)||isBlankString(titlezh)) {
                    if (orgRecord.containsKey("readme")) {
                        orgRecord.put("readmezh", orgRecord.getString("readme"));
                    }
                }
            } else {
                if (orgRecord.containsKey("readme")) {
                    orgRecord.put("readmezh", orgRecord.getString("readme"));
                }
            }

            if (orgRecord.containsKey("_key") && orgRecord.containsKey("_id")) {
                String key = orgRecord.getString("_key");
                String id = orgRecord.getString("_id");
                if (StringUtils.isNotBlank(key) && StringUtils.isNotBlank(id)) {
                    orgRecord.put("_id", id + key);
                }
            }
            //将order字段和ranking默认值改为null
            if(orgRecord.containsKey("order")&&orgRecord.containsKey("ranking")){
                Long order = orgRecord.getLong("order");
                Integer ranking = orgRecord.getInteger("ranking");
                if (order.equals(0)) {
                    orgRecord.put("order","null");
                }
                if (ranking.equals(0)) {
                    orgRecord.put("ranking","null");
                }
            }
            logger.info("数据处理的结果："+JSONObject.toJSONString(orgRecord));
        }catch (Exception e){
            e.printStackTrace();
            logger.error("处理程序异常："+e.getMessage());
        }finally {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            logger.info("单条数据处理耗时： " + (System.currentTimeMillis() - start) + " ms"+",time="+sdf.format(new Date()));
        }

    }

    public static  boolean isBlankString(String string) {
        return string == null || string.trim().isEmpty();
    }
}

